package com.hyw.trace.configuration.meta.redis.queue;

import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;

class SimpleRedisQueue extends AbstractCacheQueue {

  private static final Logger log = LoggerFactory.getLogger(SimpleRedisQueue.class);

  SimpleRedisQueue(String topic, RedisTemplate<String, String> redisTemplate, @Nullable Consumer<String> consumer) {
    super(topic, redisTemplate, consumer);
  }

  @Override
  protected void run() {
    while (isRunning()) {
      String msg = redisTemplate.opsForList().rightPop(getTopic(), timeout.getSeconds(), TimeUnit.SECONDS);
      if (Objects.nonNull(msg)) {
        try {
          this.handle(msg);
        } catch (Exception ex) {
          log.error("", ex);
        }
      }
    }
  }

  @Override
  public boolean add(String msg) {
    super.add(msg);
    redisTemplate.opsForList().leftPush(getTopic(), msg);
    return true;
  }

  @Override
  public void clear() {
    redisTemplate.delete(getTopic());
  }

}
